package com.gitee.melin.bee.core.jdbc.dialect;

import com.gitee.melin.bee.core.jdbc.ConnectionProxy;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.*;
import com.gitee.melin.bee.core.jdbc.ConnectionFactory;
import com.gitee.melin.bee.core.jdbc.DriverConnectionFactory;
import com.gitee.melin.bee.core.jdbc.QueryResult;
import com.gitee.melin.bee.util.ThreadUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.gitee.melin.bee.util.JdbcUtils;
import org.springframework.lang.Nullable;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.*;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

import static com.gitee.melin.bee.core.jdbc.enums.DataSourceType.*;
import static org.hibernate.engine.jdbc.dialect.spi.DialectResolutionInfo.NO_VERSION;

public abstract class AbstractJdbcDialect implements JdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcDialect.class);

    protected ConnectionInfo connectionInfo;

    protected ConnectionFactory connectionFactory;

    private volatile PreparedStatement statement;

    private Connection connection;

    protected String compatibleMode;

    public AbstractJdbcDialect(ConnectionInfo connectionInfo) {
        this.connectionInfo = connectionInfo;

        Map<String, String> properties = connectionInfo.getProperties();
        if (properties == null) {
            properties  = Maps.newHashMap();
        }

        connectionProperties(properties);
        connectionFactory = createConnectionFactory(properties);

        if (StringUtils.isNotBlank(connectionInfo.getCompatibleMode())) {
            compatibleMode = connectionInfo.getCompatibleMode();
        } else {
            compatibleMode = queryCompatibleMode();
        }
    }

    public AbstractJdbcDialect(Connection connection) {
        this.connection = connection;
    }

    private String queryCompatibleMode() {
        //OceanBase需判断租户信息
        String compatibleMode = "mysql";
        if (DataSourceType.OCEANBASE == connectionInfo.getDataSourceType()) {
            QueryResult queryResult = query("show variables like 'ob_compatibility_mode';");
            LinkedList<LinkedList<Object>> rows = queryResult.getRows();
            String mode = rows.get(0).get(1).toString();
            if ("oracle".equalsIgnoreCase(mode)){
                LOG.info("OceanBase compatible mode is oracle");
                compatibleMode = "oracle";
            }
        }

        return compatibleMode;
    }

    protected abstract IDBQuery getDBQuery();

    protected abstract DataTypeConvertor getTypeConvert();

    protected void connectionProperties(Map<String, String> connectionProperties) {}

    public ConnectionFactory createConnectionFactory(Map<String, String> properties) {
        try {
            return new DriverConnectionFactory(
                    getDriver(),
                    getJdbcUrl(),
                    connectionInfo,
                    properties);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    protected String getJdbcUrl() {
        return connectionInfo.getJdbcUrl();
    }

    protected abstract Driver getDriver() throws SQLException;

    protected Connection getConnection() throws SQLException {
        if (connectionFactory != null) {
            return createConnectionProxy(connectionFactory.openConnection());
        } else {
            return createConnectionProxy(connection); // 避免关闭连接
        }
    }

    protected Connection createConnectionProxy(Connection con) {
        return (Connection) Proxy.newProxyInstance(
                ConnectionProxy.class.getClassLoader(),
                new Class<?>[] {ConnectionProxy.class},
                new CloseSuppressingInvocationHandler(con));
    }

    @Override
    public MetaDataSource testConnection() {
        try (Connection connection = getConnection();
             PreparedStatement statement = connection.prepareStatement(testConnectionSql())) {
            statement.executeQuery();

            DatabaseVersion version = getDatabaseVersion(connection);

            MetaDataSource dataSourceInfo = new MetaDataSource(connection.getMetaData(), version);
            if (StringUtils.isNotBlank(compatibleMode)) {
                dataSourceInfo.setCompatibleMode(compatibleMode);
            }
            return dataSourceInfo;
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    protected String testConnectionSql() {
        return "select 1";
    }

    @Override
    public DatabaseVersion getDatabaseVersion() {
        try (Connection connection = getConnection()) {
            return getDatabaseVersion(connection);
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    private DatabaseVersion getDatabaseVersion(Connection connection) {
        try {
            DatabaseMetaData metaData = connection.getMetaData();

            try {
                return new DatabaseVersion(metaData.getDatabaseProductVersion());
            } catch (Throwable e) {
                try {
                    int major = interpretVersion( metaData.getDatabaseMajorVersion() );
                    int minor = interpretVersion( metaData.getDatabaseMinorVersion() );
                    return new DatabaseVersion(major, minor, 0);
                } catch (IllegalArgumentException e1) {
                    LOG.debug("Can't determine database version. Use default");
                    return new DatabaseVersion(0, 0, 0);
                }
            }
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    @Override
    public List<String> getDatabases() {
        try (Connection connection = getConnection();
             ResultSet resultSet = connection.getMetaData().getCatalogs()) {

            List<String> catalogs = Lists.newArrayList();
            while (resultSet.next()) {
                String catalogName = resultSet.getString(1);
                catalogs.add(catalogName);
            }
            return catalogs;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public List<String> getSchemas() {
        try (Connection connection = getConnection()) {
            return listSchemas(connection);
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected List<String> listSchemas(Connection connection) {
        try (ResultSet resultSet = connection.getMetaData().getSchemas(connection.getCatalog(), null)) {
            List<String> schemaNames = Lists.newArrayList();
            while (resultSet.next()) {
                String schemaName = resultSet.getString("TABLE_SCHEM");
                // skip internal schemas
                if (filterSchema(schemaName)) {
                    schemaNames.add(schemaName);
                }
            }
            return schemaNames;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected boolean filterSchema(String schemaName) {
        return !schemaName.equalsIgnoreCase("information_schema");
    }

    @Override
    public List<MetaTable> getSchemaTables(String schemaName) {
        return this.getSchemaTables(schemaName, null);
    }

    protected List<MetaTable> getSchemaTables(String schemaName, String tableName) {
        try (Connection connection = getConnection()) {
            String sql = getDBQuery().tablesSql(schemaName);
            if (StringUtils.isNotBlank(tableName)) {
                sql = "select * from (" + sql + ") a where "
                        + getDBQuery().tableName() + " = '" + tableName + "'";
            }

            PreparedStatement preparedStatement = null;
            ResultSet resultSet = null;
            try {
                preparedStatement = connection.prepareStatement(sql);
                if (connectionInfo != null && connectionInfo.getQueryTimeout() != null) {
                    preparedStatement.setQueryTimeout(connectionInfo.getQueryTimeout());
                }
                resultSet = preparedStatement.executeQuery();

                ResultSetMetaData metaData = resultSet.getMetaData();
                List<String> columnList = new ArrayList<>();
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    columnList.add(metaData.getColumnLabel(i));
                }

                List<MetaTable> tables = new ArrayList<>();
                while (resultSet.next()) {
                    tableName = resultSet.getString(getDBQuery().tableName());
                    if (!filterTable(tableName)) {
                        continue;
                    }
                    if (StringUtils.isNotBlank(tableName)) {
                        MetaTable tableInfo = new MetaTable();
                        tableInfo.setTableName(tableName);
                        if (columnList.contains(getDBQuery().tableComment())) {
                            tableInfo.setComment(resultSet.getString(getDBQuery().tableComment()));
                        }
                        tableInfo.setSchemaName(schemaName);
                        if (columnList.contains(getDBQuery().tableType())) {
                            String tableType = resultSet.getString(getDBQuery().tableType());
                            tableType = getTableType(tableType);
                            tableInfo.setTableType(tableType);
                        }
                        if (columnList.contains(getDBQuery().catalogName())) {
                            tableInfo.setCatalogName(resultSet.getString(getDBQuery().catalogName()));
                        }
                        if (columnList.contains(getDBQuery().engine())) {
                            tableInfo.setEngine(resultSet.getString(getDBQuery().engine()));
                        }
                        if (columnList.contains(getDBQuery().options())) {
                            tableInfo.setOptions(resultSet.getString(getDBQuery().options()));
                        }
                        if (columnList.contains(getDBQuery().rows())) {
                            tableInfo.setRows(resultSet.getLong(getDBQuery().rows()));
                        }
                        if (columnList.contains(getDBQuery().dataLength())) {
                            tableInfo.setDataLength(resultSet.getLong(getDBQuery().dataLength()));
                        }
                        if (columnList.contains(getDBQuery().createTime())) {
                            tableInfo.setCreateTime(resultSet.getTimestamp(getDBQuery().createTime()));
                        }
                        if (columnList.contains(getDBQuery().updateTime())) {
                            tableInfo.setUpdateTime(resultSet.getTimestamp(getDBQuery().updateTime()));
                        }
                        tables.add(tableInfo);
                    }
                }
                return tables;
            } finally {
                JdbcUtils.closeStatement(preparedStatement);
                JdbcUtils.closeResultSet(resultSet);
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public List<MetaColumn> getPrimaryKeys(String schemaName, String tableName) {
        return this.getSchemaColumns(schemaName, tableName)
                .stream().filter(MetaColumn::isPrimaryKey).collect(Collectors.toList());
    }

    @Override
    public MetaTable getSchemaTable(String schemaName, String tableName) {
        List<MetaTable> tables = this.getSchemaTables(schemaName, tableName);
        if (tables.isEmpty()) {
            return null;
        } else if (tables.size() > 1) {
            String msg = String.format("table %s.%s has more than one", schemaName, tableName);
            throw new RuntimeException(msg);
        } else {
            MetaTable tableInfo = tables.get(0);
            List<MetaColumn> columns = this.getSchemaColumns(schemaName, tableName);
            tableInfo.setColumns(columns);
            return tableInfo;
        }
    }

    protected boolean filterTable(String tableName) {
        return true;
    }

    @Override
    public List<MetaColumn> getSchemaColumns(String schemaName, String tableName) {
        try (Connection connection = getConnection()) {
            List<MetaColumn> columns = new ArrayList<>();

            Map<String, JDBCType> jdbcTypeMap = Maps.newHashMap();
            try {
                DatabaseMetaData metaData = connection.getMetaData();
                ResultSet results = metaData.getColumns(null, schemaName, tableName, null);
                while (results.next()) {
                    String columnName = results.getString("COLUMN_NAME");
                    int jdbcType = results.getInt("DATA_TYPE");
                    jdbcTypeMap.put(columnName, JDBCType.valueOf(jdbcType));
                }
            } catch (SQLException e) {
                throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
            }

            String sql = getDBQuery().columnsSql(schemaName, tableName);
            try (PreparedStatement preparedStatement = connection.prepareStatement(sql);
                 ResultSet resultSet = preparedStatement.executeQuery()) {

                ResultSetMetaData metaData = resultSet.getMetaData();
                List<String> columnList = new ArrayList<>();
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    columnList.add(metaData.getColumnLabel(i));
                }

                while (resultSet.next()) {
                    MetaColumn field = new MetaColumn();
                    String columnName = resultSet.getString(getDBQuery().columnName());
                    if (columnList.contains(getDBQuery().columnKey())) {
                        String key = resultSet.getString(getDBQuery().columnKey());
                        field.setPrimaryKey(StringUtils.isNotBlank(key)
                                && StringUtils.equalsIgnoreCase(getDBQuery().isPK(), key));
                    }
                    field.setColumnName(columnName);
                    field.setJdbcType(jdbcTypeMap.get(columnName));

                    if (columnList.contains(getDBQuery().columnType())) {
                        String columnType = resultSet.getString(getDBQuery().columnType());
                        if (columnType.contains("(")) {
                            String type = columnType.replaceAll("\\(.*\\)", "");
                            if (!columnType.contains(",")) {
                                Integer length = Integer.valueOf(columnType.replaceAll("\\D", ""));
                                field.setColumnLength(length);
                            } else {
                                // some database does not have precision
                                if (getDBQuery().precision() != null) {
                                    // 例如浮点类型的长度和精度是一样的，decimal(10,2)
                                    field.setColumnLength(resultSet.getInt(getDBQuery().precision()));
                                }
                            }
                            field.setColumnType(type);
                        } else {
                            field.setColumnType(columnType);
                        }
                    }

                    if (columnList.contains(getDBQuery().columnComment())
                            && StringUtils.isNotBlank(resultSet.getString(getDBQuery().columnComment()))) {
                        String columnComment =
                                resultSet.getString(getDBQuery().columnComment()).replaceAll("\"|'", "");
                        field.setComment(columnComment);
                    }
                    if (columnList.contains(getDBQuery().columnLength())) {
                        int length = resultSet.getInt(getDBQuery().columnLength());
                        if (!resultSet.wasNull()) {
                            field.setColumnLength(length);
                        }
                    }
                    if (columnList.contains(getDBQuery().isNullable())) {
                        field.setNullable(StringUtils.equalsIgnoreCase(
                                resultSet.getString(getDBQuery().isNullable()),
                                getDBQuery().nullableValue()));
                    }
                    if (columnList.contains(getDBQuery().columnPosition())) {
                        field.setPosition(resultSet.getInt(getDBQuery().columnPosition()));
                    }
                    if (columnList.contains(getDBQuery().precision())) {
                        field.setPrecision(resultSet.getInt(getDBQuery().precision()));
                    }
                    if (columnList.contains(getDBQuery().scale())) {
                        field.setScale(resultSet.getInt(getDBQuery().scale()));
                    }
                    if (columnList.contains(getDBQuery().defaultValue())) {
                        field.setDefaultValue(resultSet.getString(getDBQuery().defaultValue()));
                    }
                    if (columnList.contains(getDBQuery().autoIncrement())) {
                        field.setAutoIncrement(StringUtils.equalsIgnoreCase(
                                resultSet.getString(getDBQuery().autoIncrement()), "auto_increment"));
                    }
                    columns.add(field);
                }
            }

            return columns;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public String getCreateTableScript(String schemaName, String tableName) {
        String sql = createTableSql(schemaName, tableName);

        try (Connection connection = getConnection();
             PreparedStatement dbStat = connection.prepareStatement(sql)) {
            try (ResultSet dbResult = dbStat.executeQuery()) {
                if (dbResult.next()) {
                    String colName = getDDLColumnName();
                    return dbResult.getString(colName);
                } else {
                    return "DDL is not available";
                }
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected String createTableSql(String schemaName, String tableName) {
        return "show create table " + schemaName + "." + tableName;
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql, int maxRecords) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(sql, maxRecords), executor);
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql, int maxRecords) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(schema, sql, maxRecords), executor);
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(schema, sql, -1), executor);
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(sql, -1), executor);
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String schema, String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> execute(schema, sql), executor);
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> execute(sql), executor);
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(SchemaIdentifier schema, String sql, Object... params) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> executeUpdate(schema, sql, params), executor);
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(String sql, Object... params) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> executeUpdate(sql, params), executor);
    }

    @Override
    public QueryResult query(String sql) {
        return query(null, sql, -1);
    }

    @Override
    public QueryResult query(String schema, String sql) {
        return query(schema, sql, -1);
    }

    @Override
    public QueryResult query(String sql, int maxRecords) {
        return this.query(null, sql, maxRecords);
    }

    // https://simplefeatures.wordpress.com/2011/09/30/jdbc-tweaks-to-enable-streaming-result-sets/
    @Override
    public QueryResult query(String schema, String sql, int maxRecords) {
        String jdbcUrl = connectionInfo.getJdbcUrl();
        if (SQLSERVER == this.getDataSourceType()) {
            jdbcUrl += ";selectMethod=cursor";
        }

        try {
            Connection connection = DriverManager.getConnection(jdbcUrl,
                    connectionInfo.getUsername(), connectionInfo.getPassword());

            if (schema != null) {
                connection.setSchema(schema);
            }
            connection.setAutoCommit(false);
            try {
                statement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                if (MYSQL == this.getDataSourceType()) {
                    statement.setFetchSize(Integer.MIN_VALUE);
                } else {
                    statement.setFetchSize(100);
                }

                ResultSet result = statement.executeQuery(sql);
                QueryResult queryResult = com.gitee.melin.bee.util.JdbcUtils.handleResultSet(result, maxRecords);

                if (queryResult == null) {
                    queryResult = new QueryResult();
                }

                return queryResult;
            } finally {
                JdbcUtils.closeStatement(statement);
                JdbcUtils.closeConnection(connection);
            }
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public Boolean execute(String sql) {
        return execute(null, sql);
    }

    @Override
    public Boolean execute(String schema, String sql) {
        String jdbcUrl = connectionInfo.getJdbcUrl();

        try {
            Connection connection = DriverManager.getConnection(jdbcUrl,
                    connectionInfo.getUsername(), connectionInfo.getPassword());

            if (schema != null) {
                connection.setSchema(schema);
            }

            connection.setAutoCommit(true);
            try {
                statement = connection.prepareStatement(sql);
                return statement.execute();
            } finally {
                JdbcUtils.closeStatement(statement);
                JdbcUtils.closeConnection(connection);
            }
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public Integer executeUpdate(String sql, Object... params) {
        return this.executeUpdate(null, sql, params);
    }

    @Override
    public Integer executeUpdate(SchemaIdentifier schema, String sql, Object... params) {
        String jdbcUrl = connectionInfo.getJdbcUrl();

        try {
            Connection connection = DriverManager.getConnection(jdbcUrl,
                    connectionInfo.getUsername(), connectionInfo.getPassword());

            if (schema != null) {
                connection.setSchema(schema.getSchemaName());
            }
            connection.setAutoCommit(true);
            try {
                statement = connection.prepareStatement(sql);
                for (int i = 1; i <= params.length; i++) {
                    statement.setObject(i, params[i - 1]);
                }
                return statement.executeUpdate();
            } finally {
                JdbcUtils.closeStatement(statement);
                JdbcUtils.closeConnection(connection);
            }
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    protected String getTableType(String type) {
        if (StringUtils.containsIgnoreCase(type, "VIEW")) {
            return "VIEW";
        } else {
            return "TABLE";
        }
    }

    private int interpretVersion(int result) {
        return result < 0 ? NO_VERSION : result;
    }

    @Override
    public Statement getQueryStatement() {
        return statement;
    }

    protected abstract String getDDLColumnName();

    protected Driver loadDriver(String className) {
        try {
            Class<?> clazz = Class.forName(className);
            return (Driver) clazz.getDeclaredConstructor().newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public void close() {
        if (connectionFactory != null) {
            try {
                connectionFactory.close();
            } catch (Exception e) {}
        }
    }

    private static class CloseSuppressingInvocationHandler implements InvocationHandler {

        private final Connection target;

        public CloseSuppressingInvocationHandler(Connection target) {
            this.target = target;
        }

        @Override
        @Nullable
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            // Invocation on ConnectionProxy interface coming in...

            switch (method.getName()) {
                case "equals":
                    // Only consider equal when proxies are identical.
                    return (proxy == args[0]);
                case "hashCode":
                    // Use hashCode of PersistenceManager proxy.
                    return System.identityHashCode(proxy);
                case "close":
                    // Handle close method: suppress, not valid.
                    return null;
                case "isClosed":
                    return false;
                case "getTargetConnection":
                    // Handle getTargetConnection method: return underlying Connection.
                    return this.target;
                case "unwrap":
                    return (((Class<?>) args[0]).isInstance(proxy) ? proxy : this.target.unwrap((Class<?>) args[0]));
                case "isWrapperFor":
                    return (((Class<?>) args[0]).isInstance(proxy) || this.target.isWrapperFor((Class<?>) args[0]));
            }

            // Invoke method on target Connection.
            try {
                Object retVal = method.invoke(this.target, args);
                return retVal;
            }
            catch (InvocationTargetException ex) {
                throw ex.getTargetException();
            }
        }
    }
}
